package com.learn.code.juc.completablefuture;

import org.springframework.scheduling.concurrent.CustomizableThreadFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * @author : 翌擎科技
 * @date : 2022/8/10 11:12
 */
public class CompletableFutureCore {

    private static ThreadFactory factory = new CustomizableThreadFactory("learn_code_pool_");

    private static ThreadPoolExecutor threadPoolExecutor = new
            ThreadPoolExecutor(5, 10, 60,
            TimeUnit.SECONDS, new ArrayBlockingQueue<>(5000),
            factory,
            new ThreadPoolExecutor.AbortPolicy());

    /**
     * runAsync()是CompletableFuture最常用的方法之一，它可以接收一个待运行的任务并返回一个CompletableFuture
     */
    public static void runAsync() {
        CompletableFuture.runAsync(() -> {
            System.out.println("runAsync() 测试！" + Thread.currentThread().getName());
        }, threadPoolExecutor);
    }

    /**
     * 使用supply()时，就表明我们会返回一个结果，并且这个结果可以被后续的任务所使用。
     * 异步任务获取一个async，返回本次执行结果
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void supplyAsync() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            return "supplyAsync() 测试！" + Thread.currentThread().getName();
        }, threadPoolExecutor);
        System.out.println(future.get());
    }

    /**
     * thenApply与thenApplyAsync
     * thenApply()用于接收supply()返回的执行结果，并执行相关逻辑返回最终执行结果
     * thenApply同步
     * thenApplyAsync异步
     */

    /**
     * thenApply（）接收上一步的结果，并执行任务返回结果
     */
    public static void thenApply() throws Exception {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            return "supplyAsync() 测试！" + Thread.currentThread().getName();
        }, threadPoolExecutor).thenApply(result -> {
            return result + " thenApply() 测试！" + Thread.currentThread().getName();
        });
        System.out.println(future.get());
    }

    /**
     * thenApplyAsync（）接收上一步的结果，并执行任务返回结果
     */
    public static void thenApplyAsync() throws Exception {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            return "supplyAsync() 测试！" + Thread.currentThread().getName();
        }, threadPoolExecutor).thenApplyAsync(result -> {
            return result + " thenApplyAsync() 测试！" + Thread.currentThread().getName();
        }, threadPoolExecutor);
        System.out.println(future.get());
    }


    /**
     * * thenAccept与thenAcceptAsync
     * * thenAccep用于接收supply()返回的执行结果，并执行相关逻辑但是不返回最终执行结果，它的返回值是void
     * * thenAccept同步
     * * thenAcceptAsync异步
     */

    /**
     * thenAccept同步
     *
     * @throws Exception
     */
    public static void thenAccept() throws Exception {
        CompletableFuture.supplyAsync(() -> {
            return "supplyAsync() 测试！" + Thread.currentThread().getName();
        }, threadPoolExecutor).thenAccept(result -> {
            System.out.println(result + " thenAccept() 测试！" + Thread.currentThread().getName());
        });
    }

    /**
     * thenAcceptAsync异步
     *
     * @throws Exception
     */
    public static void thenAcceptAsync() throws Exception {
        CompletableFuture.supplyAsync(() -> {
            return "supplyAsync() 测试！" + Thread.currentThread().getName();
        }, threadPoolExecutor).thenAcceptAsync(result -> {
            System.out.println(result + " thenAcceptAsync() 测试！" + Thread.currentThread().getName());
        }, threadPoolExecutor);
    }


    /**
     * * thenCompose与 thenCombine
     * * 它们可以实现对依赖和非依赖两种类型的任务的编排。
     * * thenCompose()和thenApply()的核心不同之处在于它们的返回值类型：
     * * <p>
     * * thenApply()：返回计算结果的原始类型，比如返回String;
     * * thenCompose()：返回CompletableFuture类型，比如返回CompletableFuture.
     * * <p>
     * * thenCombine()将另外两个任务的结果同时作为参数，参与到自己的计算逻辑中。在另外两个参数未就绪时，它将会处于等待状态。
     * * <p>
     */


    public static void thenCompose() throws Exception {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            return "supplyAsync() 测试！" + Thread.currentThread().getName();
        }, threadPoolExecutor).thenCompose(result -> {
            return CompletableFuture.supplyAsync(() -> result + " thenCompose() 测试！" + Thread.currentThread().getName(), threadPoolExecutor);
        });
        System.out.println(future.get());
    }

    public static void thenComposeAsync() throws Exception {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            return "supplyAsync() 测试！" + Thread.currentThread().getName();
        }, threadPoolExecutor).thenComposeAsync(result -> {
            return CompletableFuture.supplyAsync(() -> result + " thenComposeAsync() 测试！" + Thread.currentThread().getName(), threadPoolExecutor);
        });
        System.out.println(future.get());
    }

    public static void thenCombine() throws Exception {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> " supplyAsync1() 测试！" + Thread.currentThread().getName(), threadPoolExecutor);
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> " supplyAsync2() 测试！" + Thread.currentThread().getName(), threadPoolExecutor);
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> " supplyAsync3() 测试！" + Thread.currentThread().getName(), threadPoolExecutor);


        CompletableFuture<String> combine = future.thenCombine(future2, (result1, result2) -> {
            return result1 + result2 + "combine end " + Thread.currentThread().getName();
        });
        System.out.println(combine.get());
    }

    public static void thenCombineAsync() throws Exception {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> " supplyAsync1() 测试！" + Thread.currentThread().getName(), threadPoolExecutor);
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> " supplyAsync2() 测试！" + Thread.currentThread().getName(), threadPoolExecutor);
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> " supplyAsync3() 测试！" + Thread.currentThread().getName(), threadPoolExecutor);


        CompletableFuture<String> combine = future.thenCombineAsync(future2, (result1, result2) -> {
            return result1 + result2 + "    combine end " + Thread.currentThread().getName();
        }, threadPoolExecutor);
        System.out.println(combine.get());
    }


    /**
     * allOf()与anyOf()
     * 用于对多个Future进行组织 相当于CountDownLatch
     * allOf()：给定一组任务，等待所有任务执行结束；
     * anyOf()：给定一组任务，等待其中任一任务执行结束。
     * anyOf()将返回完任务的执行结果，但是allOf()不会返回任何结果，它的返回值是Void.
     */
    public static void allOf() throws Exception {
        CompletableFuture<String> any1 = CompletableFuture.supplyAsync(() -> {
            String s = "第1个任务！" + Thread.currentThread().getName();
            System.out.println(s);
            return s;
        }, threadPoolExecutor);
        CompletableFuture<String> any2 = CompletableFuture.supplyAsync(() -> {
            String s = "第2个任务！" + Thread.currentThread().getName();
            System.out.println(s);
            return s;
        }, threadPoolExecutor);
        CompletableFuture<String> any3 = CompletableFuture.supplyAsync(() -> {
            String s = "第3个任务！" + Thread.currentThread().getName();
            System.out.println(s);
            return s;
        }, threadPoolExecutor);
        CompletableFuture<String> any4 = CompletableFuture.supplyAsync(() -> {
            String s = "第4个任务！" + Thread.currentThread().getName();
            System.out.println(s);
            return s;
        }, threadPoolExecutor);
        CompletableFuture<String> any5 = CompletableFuture.supplyAsync(() -> {
            String s = "第5个任务！" + Thread.currentThread().getName();
            System.out.println(s);
            return s;
        }, threadPoolExecutor);

        CompletableFuture<Void> future3 = CompletableFuture.allOf(any1, any2, any3, any4, any5);
        System.out.println(future3.get());

        CompletableFuture.runAsync(() -> System.out.println("第end个任务！" + Thread.currentThread().getName()), threadPoolExecutor);

    }

    public static void allOfForEach() throws Exception {
        List<CompletableFuture<String>> list = new ArrayList<>();
        for (int i = 0; i < 1000; i++) {
            int finalI = i;
            CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
                String s = "第" + finalI + "个任务！" + Thread.currentThread().getName();
                System.out.println(s);
                return s;
            }, threadPoolExecutor);
            list.add(future);
        }
        CompletableFuture[] futures = list.toArray(new CompletableFuture[0]);
        CompletableFuture.allOf(futures).thenRunAsync(() -> {
            System.out.println("阻塞");
        }).get();

        CompletableFuture.runAsync(() -> {
            System.out.println("阻塞后执行");
        });
    }

    public static void anyOf() throws Exception {
        CompletableFuture<String> any1 = CompletableFuture.supplyAsync(() -> {
            String s = "第1个任务！" + Thread.currentThread().getName();
            System.out.println(s);
            return s;
        }, threadPoolExecutor);
        CompletableFuture<String> any2 = CompletableFuture.supplyAsync(() -> {
            String s = "第2个任务！" + Thread.currentThread().getName();
            System.out.println(s);
            return s;
        }, threadPoolExecutor);
        CompletableFuture<String> any3 = CompletableFuture.supplyAsync(() -> {
            String s = "第3个任务！" + Thread.currentThread().getName();
            System.out.println(s);
            return s;
        }, threadPoolExecutor);
        CompletableFuture<String> any4 = CompletableFuture.supplyAsync(() -> {
            String s = "第4个任务！" + Thread.currentThread().getName();
            System.out.println(s);
            return s;
        }, threadPoolExecutor);
        CompletableFuture<String> any5 = CompletableFuture.supplyAsync(() -> {
            String s = "第5个任务！" + Thread.currentThread().getName();
            System.out.println(s);
            return s;
        }, threadPoolExecutor);

        CompletableFuture<Object> future3 = CompletableFuture.anyOf(any1, any2, any3, any4, any5);
        System.out.println(future3.get());

        CompletableFuture.runAsync(() -> System.out.println("第end个任务！" + Thread.currentThread().getName()), threadPoolExecutor);

    }

    /**
     * CompletableFuture中的异常处理
     * 在CompletableFuture链式调用中，如果某个任务发生了异常，那么后续的任务将都不会再执行。对于异常，我们有两种处理方式：exceptionally()和handle().
     * 1. 使用exceptionally()回调处理异常
     * 2. 使用handle()处理异常
     */

    /**
     * 发生异常才会执行exceptionally
     *
     * @throws Exception
     */
    public static void exceptionally() throws Exception {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            String s = "第1个任务！" + Thread.currentThread().getName();
            System.out.println(s);
            return s;
        });
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            int a = 1 / 0;
            String s = "第2个任务！" + Thread.currentThread().getName();
            System.out.println(s);
            return s;
        });
        CompletableFuture<String> future = future1.thenCombine(future2, (r1, r2) -> {
            return r1 + r2 + "end";
        }).exceptionally(e -> {
            System.out.println("发生异常信息：" + e.toString());
            return e.toString();
        });
        System.out.println(future.get());
    }

    /**
     * handle 不论是是否发生异常都会执行handle
     *
     * @throws Exception
     */
    public static void handle() throws Exception {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            String s = "第1个任务！" + Thread.currentThread().getName();
            System.out.println(s);
            return s;
        });
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            int a = 1 / 0;
            String s = "第2个任务！" + Thread.currentThread().getName();
            System.out.println(s);
            return s;
        });
        CompletableFuture<String> future = future1.thenCombine(future2, (r1, r2) -> {
            return r1 + r2 + "end";
        }).handle((res, e) -> {
            System.out.println("执行结果------");
            if (e != null) {
                System.out.println("发生异常+" + e.toString());
                return "";
            }
            return res;
        });
        System.out.println(future.get());
    }


    public static void main(String[] args) throws Exception {
        runAsync();
        System.out.println("-----------------------------------------------------------------");
        supplyAsync();
        System.out.println("-----------------------------------------------------------------");
        thenApply();
        System.out.println("-----------------------------------------------------------------");
        thenApplyAsync();
        System.out.println("-----------------------------------------------------------------");
        thenAccept();
        System.out.println("-----------------------------------------------------------------");
        thenAcceptAsync();
        System.out.println("-----------------------------------------------------------------");
        thenCompose();
        System.out.println("-----------------------------------------------------------------");
        thenComposeAsync();
        System.out.println("-----------------------------------------------------------------");
        thenCombine();
        System.out.println("-----------------------------------------------------------------");
        thenCombineAsync();
        System.out.println("-----------------------------------------------------------------");
        allOf();
        System.out.println("-----------------------------------------------------------------");
        allOfForEach();
        System.out.println("-----------------------------------------------------------------");
        anyOf();
        System.out.println("-----------------------------------------------------------------");
        exceptionally();
        System.out.println("-----------------------------------------------------------------");
        handle();
    }
}
